﻿#include "TcpConnection.h"
#include "as_channel.h"
#include "byte_buffer.h"
#include "as_awaiter.h"
#include "log.h"
#include "cross_comm.h"
#include <iostream>

as::net::TcpConnection::TcpConnection(std::shared_ptr<EpollWatch> &ep, socket_fd fd, const std::string &name, Address &localAddr, Address &peerAddr)
	:ep_(ep), local_addr_(localAddr), peer_addr_(peerAddr)
{
	state_ = CONNECTING;
	s_ = std::make_unique<Socket>(fd);
	s_->set_non_block();
	s_->set_recv_buff(E_RECV_SIZE);
	s_->set_recv_buff(E_WRITE_SIZE);
	recv_buf_ = std::make_shared<ByteBuffer>();
	recv_buf_->make_writeable_bytes(E_RECV_SIZE / 2);
	write_buf_ = std::make_shared<ByteBuffer>();
	write_buf_->make_writeable_bytes(E_WRITE_SIZE / 2);

	channel_ = std::make_shared<Channel>(ep, fd);
	channel_->SetReadCallback(std::bind(&TcpConnection::HandleRead, this, std::placeholders::_1));
	channel_->SetWriteCallback(std::bind(&TcpConnection::HandleWrite, this));
	channel_->SetCloseCallback(std::bind(&TcpConnection::HandleClose, this));
	channel_->SetErrorCallback(std::bind(&TcpConnection::HandleError, this));
}

as::net::TcpConnection::~TcpConnection()
{
	LOG_DEBUG << "delete tcpconnection:" << this->GetConnId() << " fd:" << this->GetFd();
}

void as::net::TcpConnection::Shutdown()
{
	state_ = DISCONNECT;
	if (channel_->IsWriteFlag())
	{
		s_->shutdown(AS_SD_SEND);
	}
}

void as::net::TcpConnection::SendMsg(const char *buf, int size)
{
	if (!channel_->IsWriteFlag() && write_buf_->ReadableBytes() == 0)
	{
		int sendSize = Socket::write(s_->get_fd(), buf, size);
		LOG_DEBUG << "tcp connection send msg, connid:" << GetConnId() << " fd:" << s_->get_fd() << " size:" << size << " sendSize:" << sendSize;
		if (sendSize <= 0)
		{
			LOG_WARNING << "send data is error:" << sendSize;
		}
		if (sendSize < size)
		{
			write_buf_->write_bytes((uint8*)buf + sendSize, size - sendSize);
			channel_->EnableWriteFlag();
		}
	}
	else
	{
		LOG_DEBUG << "tcp connection send msg add to wait, connid:" << GetConnId() << " fd:" << s_->get_fd() << " size:" << size;
		write_buf_->write_bytes((uint8*)buf, size);
		if (!channel_->IsWriteFlag())
		{
			channel_->EnableWriteFlag();
		}
	}
}

void as::net::TcpConnection::SendMsg(ByteBuffer &buf)
{
	SendMsg((const char*)buf.GetReadPtr(), buf.ReadableBytes());
}

void as::net::TcpConnection::SendMsg(ByteBufferPtr buf)
{
	SendMsg((const char*)buf->GetReadPtr(), buf->ReadableBytes());
}

void as::net::TcpConnection::SetMsgCallback(MsgCallback fun)
{
	msg_cb_ = fun;
}

void as::net::TcpConnection::SetConnectCallback(ConnChangeCb fun)
{
	conn_change_cb_ = fun;
}

void as::net::TcpConnection::SetCloseCallback(CloseCallback fun)
{
	close_cb_ = fun;
}

void as::net::TcpConnection::SetConnId(int id)
{
	conn_id_ = id;
}

std::shared_ptr<as::net::EpollWatch> as::net::TcpConnection::GetWatch()
{
	return ep_.expired() == false ? ep_.lock() : nullptr;
}

as::co::ReadAwaiter as::net::TcpConnection::co_read()
{
	return { .conn_ = this->shared_from_this(), .channel_ = channel_ };
}

int as::net::TcpConnection::HandleRead(TimeValue &tv)
{
	int totalReadCount = 0;
	int readn = 0;
	do 
	{
		int canWriteCount = recv_buf_->WriteableBytes();
		if (canWriteCount < 10240)
		{
			recv_buf_->gc();
			canWriteCount = recv_buf_->WriteableBytes();
			if (canWriteCount < 10240)
			{
				HandleError();
				return -1;
			}
		}

		uint8 *buf = recv_buf_->GetWritePtr();
		readn = Socket::recv(s_->get_fd(), buf, canWriteCount);
		if(readn > 0)
			totalReadCount += readn;
		HandleReadAfter(readn);
	} while (readn > 0);
	return totalReadCount;
}

void as::net::TcpConnection::HandleReadAfter(int readn)
{
	LOG_DEBUG << "tcp connect recv size:" << readn << " connid:" << this->GetConnId() << " fd:" << s_->get_fd();
	if (readn > 0)
	{
		if (msg_cb_)
		{
			recv_buf_->AddWritePos(readn);
			auto self = shared_from_this();
			msg_cb_(self, recv_buf_, readn);
		}
	}
	else if (readn == 0)
	{
		auto err = as::tools::GetLastError();
		LOG_WARNING << "connection recv is close, connid:" << this->GetConnId() << " fd:" << s_->get_fd();
		HandleClose();
	}
	else
	{
		if (readn == -1)
		{
			auto err = as::tools::GetLastError();
#ifdef _WIN32
			if (err == WSAEWOULDBLOCK)
			{
				return;
			}
#endif
		}

		LOG_WARNING << "connection recv is error, readn:" << readn << " connid:" << this->GetConnId() << " fd : " << s_->get_fd();
		HandleError();
	}
}

void as::net::TcpConnection::HandleWrite()
{
	if (channel_->IsWriteFlag())
	{
		if (write_buf_->ReadableBytes() <= 0)
		{
			//设置读状态
			channel_->DisableWriteFlag();
			return;
		}

		int n = Socket::write(channel_->GetFd(),
			write_buf_->GetReadPtr(), write_buf_->ReadableBytes());

		if (n <= 0)
		{
			LOG_WARNING << "HandleWrite error:" << n << " fd:" << channel_->GetFd();
			return;
		}

		write_buf_->AddReadPos(n);
		//这里缓冲区已经读完
		if (write_buf_->ReadableBytes() == 0)
		{
			//设置读状态
			channel_->DisableWriteFlag();
		}
	}
}

void as::net::TcpConnection::HandleClose()
{
	state_ = DISCONNECT;
	channel_->ClearFlag();

	if (conn_change_cb_)
	{
		conn_change_cb_(shared_from_this());
	}

	if (close_cb_)
	{
		close_cb_(shared_from_this());
	}

}

void as::net::TcpConnection::HandleError()
{
	state_ = DISCONNECT;
	channel_->ClearFlag();

	if (err_cb_)
	{
		err_cb_(shared_from_this(), s_->get_last_err());
	}
}

void as::net::TcpConnection::ConnectFinish()
{
	if (state_ == CONNECTING)
	{
		state_ = CONNECTED;
		channel_->EnableReadFlag();
		if (conn_change_cb_)
		{
			conn_change_cb_(shared_from_this());
		}
	}
}

void as::net::TcpConnection::DestoryConn()
{
	if (state_ == CONNECTED)
	{
		state_ = DISCONNECT;
		channel_->ClearFlag();
		if (conn_change_cb_)
		{
			conn_change_cb_(shared_from_this());
		}
	}
	channel_->RemoveLoop();

}
